#!/bin/sh
###############################################################################
##  Author    : xuezhoyi
##  Name      : edw_data_exporter.sh
##  Functions : 数据导出
##  Purpose   : 
##  Revisions or Comments
##  VER        DATE        AUTHOR           DESCRIPTION
##---------  ----------  ---------------  ------------------------------------ 
##  1.0      2017-05-22  xuezhouyi        1. CREATED THIS SHELL.
##  2.0      2018-11-11  薛周毅           1.根据入参判断是hive2mysql还是mysql2hive
##                                        2.根据表【EDW_DATA_SYNC】开始数据导出
###############################################################################

. ${0%`basename ${0}`}edw_parm.sh
. ${0%`basename ${0}`}edw_func.sh

function USAGE(){
    echo "How to use this shell script!"
    echo "${V_SHELL_HOME}/edw_data_exporter.sh M2H 1"
}

# ensure the basic variavle of connection is not null
if [[ -z ${1} || -z ${2} ]];then
    USAGE
    exit 1
else
    V_MODE=${1}
    V_PARALLEL_CNT=${2}
fi
V_CURR_DATE=$(grep "#V_CURR_DATE#" ${V_PARM_FILE} | awk -F'=' '{print $2}')
V_DATA_DATE=$(grep "#V_DATA_DATE#" ${V_PARM_FILE} | awk -F'=' '{print $2}')
V_FREQUENCY=$(grep "#V_FREQUENCY#" ${V_PARM_FILE} | awk -F'=' '{print $2}')
# Define basic parm
V_SCRIPT_NAME=$(basename ${0} | cut -d'.' -f1)
V_OUT_LOGS="${V_SHELL_LOGS}/${V_DATA_DATE}/${V_SCRIPT_NAME}_${V_FREQUENCY}_${V_PARALLEL_CNT}.log"

# Define the output flow
exec 4>&1               # screen output
exec 3>>${V_OUT_LOGS}   # script output
exec 2>&3               # error output
exec 1>&3               # standard output



BEGIN
LOGGER "INFO" "[------------------------Thread:${V_PARALLEL_CNT}------------------------]"
LOGGER "INFO" "Today batch date is ${V_DATA_DATE}."

# Get table list
V_QUERY="SELECT TABLE_NAME FROM ETL.EDW_DATA_SYNC WHERE SYNC_TYPE='${V_MODE}' AND MOD(SEQ_NO,${V_PARALLEL_SYNC_NO})='${V_PARALLEL_CNT}' AND IF_VALID='Y'"
V_TAB_LIST=$(${V_RUN_DB} -q "${V_QUERY}") || exit 2
for TAB in ${V_TAB_LIST};do
    LOGGER "INFO" "表[${TAB}]开始同步. "
    # Get table column
    case ${V_DB_TYPE} in
      mysql)
          V_QUERY="SELECT GROUP_CONCAT(B.COLUMN_NAME) FROM ETL.EDW_DATA_SYNC A INNER JOIN INFORMATION_SCHEMA.COLUMNS B ON UPPER(A.TABLE_NAME)=UPPER(CONCAT(B.TABLE_SCHEMA,'.',B.TABLE_NAME)) WHERE A.TABLE_NAME='${TAB}' AND A.SYNC_TYPE='${V_MODE}' AND SYNC_FREQUENCY='${REQ}' GROUP BY A.TABLE_NAME ORDER BY B.ORDINAL_POSITION;"
      ;;
      db2)
          V_QUERY="SELECT B.NAME FROM ETL.EDW_DATA_SYNC A INNER JOIN SYSIBM.SYSCOLUMNS B ON CONCAT(RTRIM(B.TBCREATOR),CONCAT('.',B.TBNAME)) = UPPER(A.TABLE_NAME) WHERE A.TABLE_NAME='${TAB}' AND A.SYNC_TYPE='${V_MODE}'  ORDER BY B.COLNO";
      ;;
    esac
    V_TAB_COL_LIST=$(${V_RUN_DB} -q "${V_QUERY}") || exit 3
    i=0
    for COL in ${V_TAB_COL_LIST};do
      i=$(( $i + 1 ))
      if [ ${i} = 1 ];then
        V_TAB_COL=${COL}
      else
        V_TAB_COL=${V_TAB_COL}","${COL}
      fi
    done
    # Get where condition
    V_QUERY="SELECT COALESCE(WHERE_CONDITION,'') FROM ETL.EDW_DATA_SYNC WHERE TABLE_NAME='${TAB}' AND SYNC_TYPE='${V_MODE}'"
    V_WHERE=$(${V_RUN_DB} -q "${V_QUERY}") || exit 4
    if [[ ${V_WHERE}x == ''x ]];then
        V_QUERY="SELECT ${V_TAB_COL} FROM ${TAB}"
    else #???
        V_QUERY=$(eval echo '$'"XAVIER SELECT ${V_TAB_COL} FROM ${TAB} WHERE ${V_WHERE}")
        RE_CODE=$?
        if [[ ${RE_CODE} -ne 0 ]]; then
            LOGGER "ERROR 5" "查询条件替换参数失败"
            LOGGER "DEBUG" "exec=\$(eval echo '\$'\"XAVIER SELECT ${V_TAB_COL} FROM ${TAB} WHERE ${V_WHERE}\"),V_DATA_DATE=\"${V_DATA_DATE}\""
            exit 5
        fi
    fi
    
    # Target output file
    V_FILE="${V_SHELL_DATA}/${V_DATA_DATE}/${TAB}.TXT"
    
    ## 数据导出
    # swith case
    case ${V_MODE} in
        M2H)
            # Export data
            ${V_RUN_DB} -x "export to ${V_FILE} of del ${V_QUERY}" || exit 6
        ;;
        H2M)
            # Export data
            ${V_RUN_HIVQL} -x "${V_QUERY}" > "${V_FILE}" || exit 7
        ;;
    esac

    sed -i s/\"//g "${V_FILE}" || exit 8
    sed -i s/,/${V_DELIMITER}/g "${V_FILE}" || exit 9
    
    ## 数据导入
    # swith case
    case ${V_MODE} in
        M2H)
            V_QUERY="LOAD DATA LOCAL INPATH '${V_FILE}' OVERWRITE INTO TABLE ${TAB}"
            ${V_RUN_HIVQL} -q "${V_QUERY}" || exit 4
        ;;
        H2M)
            # Get where condition
            if [[ ${V_WHERE} == '' ]];then
                V_QUERY="TRUNCATE TABLE ${TAB}"
            else #???
                V_QUERY=$(eval echo '$'"XAVIER DELETE FROM ${TAB} WHERE ${V_WHERE}")
                RE_CODE=$?
                if [[ ${RE_CODE} -ne 0 ]]; then
                    LOGGER "ERROR" "查询条件替换参数失败"
                    LOGGER "DEBUG" "exec=\$(eval echo '\$'\"XAVIER SELECT ${V_TAB_COL} FROM ${TAB} WHERE ${V_WHERE}\"),V_DATA_DATE=\"${V_DATA_DATE}\""
                    exit 5
                fi
            fi
            # Load data before delete
            ${V_RUN_DB} -q "${V_QUERY}" || exit 6
            case ${V_DB_TYPE} in
              mysql)
                  V_QUERY="LOAD DATA LOCAL INFILE '${V_FILE}' INTO TABLE ${TAB} CHARACTER SET UTF8 FIELDS TERMINATED BY '${V_DELIMITER}'"
              ;;
              db2)
                  sed -i s/${V_DELIMITER}/,/g "${V_FILE}" || exit 7
                  V_QUERY="import from '${V_FILE}' of del insert into ${TAB}"
              ;;
            esac
            ${V_RUN_DB} -q "${V_QUERY}" || exit 8
            
        ;;
    esac
    LOGGER "INFO" "表[${TAB}] 同步成功. "
    echo "TAB_EXP=1">>${SYNC_SPEED_FILE}
done

END
exit 0
